/**
 * @fileName: DefRocketConsumerImpl.java
 * @author: pjf
 * @date: 2019年10月17日 下午3:19:02
 */

package com.rgs.base.rocketmq.def;


import com.rgs.base.rocketmq.RocketMqConsumer;
import com.rgs.base.rocketmq.annotation.RocketMqConsumerListenAnnotation;
import com.rgs.base.rocketmq.config.RocketMqSource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.function.Consumer;

/**
 * @author pjf
 * @className DefRocketConsumerImpl.java
 * @date 2019年10月17日 下午3:19:02
 */
@Slf4j
public class DefRocketConsumerImpl implements RocketMqConsumer {
    private DefaultMQPushConsumer consumer;

    private Consumer<MessageExt> consumerHandle;
    //    private String topic;
//    private String subExpression;
//    private MessageModel messageModel = MessageModel.CLUSTERING;
    private boolean start = false;
    private RocketMqConsumerListenAnnotation rocketMqConsumerListenAnnotation;

    public DefRocketConsumerImpl(RocketMqSource rocketMqSource) {
        // 生成Consumer
        consumer = new DefaultMQPushConsumer(rocketMqSource.getRocketGroup());
        // 配置Consumer
        consumer.setConsumeMessageBatchMaxSize(32);
        consumer.setNamesrvAddr(rocketMqSource.getRocketServers());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //
    }

    @Override
    public void start() {
        consumer.registerMessageListener((MessageListenerConcurrently)
                (list, consumeConcurrentlyContext) -> {
                    // 消费消息
                    for (MessageExt me : list) {
                        try {
                            consumerHandle.accept(me);
                        } catch (Exception ex) {
                            log.error(ex.getMessage());
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                });
        // 启动Consumer
        try {
//            consumer.setMessageModel(messageModel);
//            consumer.subscribe(topic, subExpression);
            consumer.setMessageModel(rocketMqConsumerListenAnnotation.messageModel());
            consumer.subscribe(rocketMqConsumerListenAnnotation.topic(), rocketMqConsumerListenAnnotation.subExpression());
            consumer.start();
            start = true;
        } catch (MQClientException e) {
            log.error(e.getMessage());
        }
    }

    @Override
    public void disponse() {
        consumer.shutdown();
        start = false;
    }

    @Override
    public void setConsumerHandle(Consumer<MessageExt> consumerHandle) {
        this.consumerHandle = consumerHandle;

    }

    @Override
    public void setRocketMqConsumerAnnotation(RocketMqConsumerListenAnnotation rocketMqConsumerListenAnnotation) {
        this.rocketMqConsumerListenAnnotation = rocketMqConsumerListenAnnotation;
    }

//    @Override
//    public void setTopicAndSubExpression(String topic, String subExpression) {
//        this.topic = topic;
//        this.subExpression = subExpression;
//    }
//
//    @Override
//    public void setMessageModel(MessageModel messageModel) {
//        this.messageModel = messageModel;
//    }

    @Override
    public boolean isStarted() {
        return start;
    }

//	@Override
//	public void setSuccessTopic(String successTopic) {
//		this.successTopic = successTopic;
//	}
//
//	@Override
//	public void setErrorTopic(String errorTopic) {
//		this.errorTopic = errorTopic;
//	}
}
